Helper: simulate typing a command at a shell prompt, then run it

run_cmd() {
echo ""
echo -e "\033[1;32m$\033[0m $1"
sleep 0.3
eval "$1"
}

── 1. simple-tool ──────────────────────────────────────────────────────────

echo "Recording 01-simple-tool..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-tool example"
--command 'bash -c "
run_cmd() { echo ""; echo -e "\033[1;32m\$\033[0m $1"; sleep 0.3; eval "$1"; }

echo "━━━ simple-tool (stdio) ━━━"
echo "A server with a \"fetch\" tool that fetches website content."
echo ""

run_cmd "cd examples/servers/simple-tool"

cat << "SCRIPT"

$ uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\",
args=[\"run\", \"mcp-simple-tool\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
for t in tools.tools:
print(f\" Tool: {t.name}\")
print(f\" Description: {t.description}\")
anyio.run(main)
"
SCRIPT

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-tool\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
for t in tools.tools:
print(f\" Tool: {t.name}\")
print(f\" Description: {t.description}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/01-simple-tool.cast 2>&1 | tail -1

── 2. simple-prompt ────────────────────────────────────────────────────────

echo "Recording 02-simple-prompt..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-prompt example"
--command 'bash -c "
echo "━━━ simple-prompt (stdio) ━━━"
echo "A server with a \"simple\" prompt that takes optional context and topic."
echo ""

echo -e "\033[1;32m$\033[0m cd examples/servers/simple-prompt"
cd examples/servers/simple-prompt

cat << "SCRIPT"

$ uv run python -c "...client script..."
SCRIPT

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-prompt\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
prompts = await session.list_prompts()
for p in prompts.prompts:
print(f\" Prompt: {p.name} - {p.description}\")
print()
result = await session.get_prompt(\"simple\", {\"context\": \"testing kwargs\", \"topic\": \"MCP SDK\"})
for msg in result.messages:
print(f\" [{msg.role}]: {msg.content.text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/02-simple-prompt.cast 2>&1 | tail -1

── 3. simple-resource ──────────────────────────────────────────────────────

echo "Recording 03-simple-resource..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-resource example"
--command 'bash -c "
echo "━━━ simple-resource (stdio) ━━━"
echo "A server with 3 text resources: greeting, help, about."
echo ""
echo -e "\033[1;32m$\033[0m cd examples/servers/simple-resource"
cd examples/servers/simple-resource
echo -e "\033[1;32m$\033[0m uv run python -c \"...client...\""
echo ""
uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-resource\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
resources = await session.list_resources()
print(f\" Found {len(resources.resources)} resources:\")
for r in resources.resources:
result = await session.read_resource(r.uri)
print(f\" {r.name}: {result.contents[0].text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/03-simple-resource.cast 2>&1 | tail -1

── 4. simple-pagination ────────────────────────────────────────────────────

echo "Recording 04-simple-pagination..."
asciinema rec --overwrite --idle-time-limit 2 --title "simple-pagination example"
--command 'bash -c "
echo "━━━ simple-pagination (stdio) ━━━"
echo "25 tools (5/page), 30 resources (10/page), 20 prompts (7/page)."
echo ""
echo -e "\033[1;32m$\033[0m cd examples/servers/simple-pagination"
cd examples/servers/simple-pagination
echo -e "\033[1;32m$\033[0m uv run python -c \"...client...\""
echo ""
uv run python -c "
import anyio
from mcp import types
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(command=\"uv\", args=[\"run\", \"mcp-simple-pagination\", \"--transport\", \"stdio\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
all_tools, cursor, page = [], None, 0
while True:
p = types.PaginatedRequestParams(cursor=cursor) if cursor else None
result = await session.list_tools(params=p)
all_tools.extend(result.tools)
page += 1
print(f\" Page {page}: {len(result.tools)} tools (next_cursor={result.next_cursor})\")
if not result.next_cursor: break
cursor = result.next_cursor
print(f\" Total: {len(all_tools)} tools across {page} pages\")
print()
result = await session.call_tool(\"tool_1\", {\"input\": \"hello\"})
print(f\" call_tool result: {result.content[0].text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/04-simple-pagination.cast 2>&1 | tail -1

── 5-8. Snippets ───────────────────────────────────────────────────────────

echo "Recording 05-snippet-basic..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: basic.py"
--command 'bash -c "
echo "━━━ snippet: lowlevel/basic.py (stdio) ━━━"
echo "Basic low-level server with prompts using kwargs constructor."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/basic.py"
echo " (server starts via stdio, client connects automatically)"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/basic.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
prompts = await session.list_prompts()
print(f\" Prompts: {[p.name for p in prompts.prompts]}\")
result = await session.get_prompt(\"example-prompt\", {\"arg1\": \"test-value\"})
print(f\" Message: {result.messages[0].content.text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/05-snippet-basic.cast 2>&1 | tail -1

echo "Recording 06-snippet-direct-call-tool..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: direct_call_tool_result.py"
--command 'bash -c "
echo "━━━ snippet: direct_call_tool_result.py (stdio) ━━━"
echo "Tool returning CallToolResult with structured_content and _meta."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/direct_call_tool_result.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/direct_call_tool_result.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
result = await session.call_tool(\"advanced_tool\", {\"message\": \"hello\"})
print(f\" Text: {result.content[0].text}\")
print(f\" Structured: {result.structured_content}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/06-snippet-direct-call-tool.cast 2>&1 | tail -1

echo "Recording 07-snippet-lifespan..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: lifespan.py"
--command 'bash -c "
echo "━━━ snippet: lifespan.py (stdio) ━━━"
echo "Server with lifespan: database connects on start, disconnects on stop."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/lifespan.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/lifespan.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
result = await session.call_tool(\"query_db\", {\"query\": \"SELECT * FROM users\"})
print(f\" Result: {result.content[0].text}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/07-snippet-lifespan.cast 2>&1 | tail -1

echo "Recording 08-snippet-structured-output..."
asciinema rec --overwrite --idle-time-limit 2 --title "snippet: structured_output.py"
--command 'bash -c "
echo "━━━ snippet: structured_output.py (stdio) ━━━"
echo "Tool with output_schema and structured_content response."
echo ""
echo -e "\033[1;32m$\033[0m uv run --frozen python examples/snippets/servers/lowlevel/structured_output.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\", args=[\"run\", \"--frozen\", \"python\", \"examples/snippets/servers/lowlevel/structured_output.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
print(f\" Tool: {tools.tools[0].name}\")
print(f\" Has output_schema: {tools.tools[0].output_schema is not None}\")
result = await session.call_tool(\"get_weather\", {\"city\": \"San Francisco\"})
print(f\" Structured: {result.structured_content}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/08-snippet-structured-output.cast 2>&1 | tail -1

── 9. structured-output-lowlevel ───────────────────────────────────────────

echo "Recording 09-structured-output-lowlevel..."
asciinema rec --overwrite --idle-time-limit 2 --title "structured-output-lowlevel example"
--command 'bash -c "
echo "━━━ structured-output-lowlevel (stdio) ━━━"
echo "Full server example with structured output (simulated weather)."
echo ""
echo -e "\033[1;32m$\033[0m python examples/servers/structured-output-lowlevel/mcp_structured_output_lowlevel/main.py"
echo ""
uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
params = StdioServerParameters(
command=\"uv\",
args=[\"run\", \"--frozen\", \"python\",
\"examples/servers/structured-output-lowlevel/mcp_structured_output_lowlevel/main.py\"])
async with stdio_client(params) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
print(f\" Tool: {tools.tools[0].name} (output_schema: {tools.tools[0].output_schema is not None})\")
result = await session.call_tool(\"get_weather\", {\"city\": \"Tokyo\"})
print(f\" Structured: {result.structured_content}\")
anyio.run(main)
"
echo ""
echo "✓ Done"
"' evidence/09-structured-output-lowlevel.cast 2>&1 | tail -1

── 10-15. HTTP servers ──────────────────────────────────────────────────────

record_http_demo() {
local num="$1"
local name="$2"
local title="$3"
local desc="$4"
local server_dir="$5"
local server_cmd="$6"
local port="$7"
local client_code="$8"

echo "Recording ${num}-${name}..."
asciinema rec --overwrite --idle-time-limit 2 --title "$title" \
  --command "bash -c \"

echo \"━━━ $title ━━━\"
echo \"$desc\"
echo \"\"
echo -e \"\\033[1;34m# Terminal 1: Start server\\033[0m\"
echo -e \"\\033[1;32m\\$\\033[0m cd $server_dir\"
echo -e \"\\033[1;32m\\$\\033[0m $server_cmd\"
echo \"\"

cd $server_dir
$server_cmd &
SERVER_PID=\$!
cd /fake/path/python-sdk

for i in \$(seq 1 30); do
code=\$(curl -s -o /dev/null -w \"%{http_code}\" http://127.0.0.1:$port/mcp 2>/dev/null || echo \"000\")
if echo \"\$code\" | grep -qE \"(200|307|400|405)\"; then break; fi
sleep 0.5
done

echo \" Server is running on http://127.0.0.1:$port\\\"
echo \"\"
echo -e \"\\033[1;34m# Terminal 2: Connect client\\033[0m\"
echo -e \"\\033[1;32m\\$\\033[0m uv run --frozen python -c \\\"...client...\\\"\"
echo \"\"

uv run --frozen python -c \"$client_code\"

echo \"\"
echo \"✓ Done\"

kill \$SERVER_PID 2>/dev/null; wait \$SERVER_PID 2>/dev/null
"" "evidence/${num}-${name}.cast" 2>&1 | tail -1
}

record_http_demo "10" "simple-streamablehttp"
"simple-streamablehttp (HTTP)"
"Streamable HTTP with notification stream and event store."
"examples/servers/simple-streamablehttp"
"uv run mcp-simple-streamablehttp --port 23000"
"23000"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23000/mcp') as client:
tools = await client.list_tools()
print(f' Tools: {[t.name for t in tools.tools]}')
result = await client.call_tool('start-notification-stream', {'interval': 0.1, 'count': 3, 'caller': 'demo'})
print(f' Result: {result.content[0].text}')

anyio.run(main)
"

record_http_demo "11" "simple-streamablehttp-stateless"
"simple-streamablehttp-stateless (HTTP)"
"Stateless streamable HTTP (no session persistence)."
"examples/servers/simple-streamablehttp-stateless"
"uv run mcp-simple-streamablehttp-stateless --port 23001"
"23001"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23001/mcp') as client:
tools = await client.list_tools()
print(f' Tools: {[t.name for t in tools.tools]}')
result = await client.call_tool('start-notification-stream', {'interval': 0.1, 'count': 3, 'caller': 'demo'})
print(f' Result: {result.content[0].text}')

anyio.run(main)
"

record_http_demo "12" "everything-server"
"everything-server (HTTP)"
"Conformance server: 13 tools, resources, prompts, completions."
"examples/servers/everything-server"
"uv run mcp-everything-server --port 23002"
"23002"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23002/mcp') as client:
tools = await client.list_tools()
print(f' Tools ({len(tools.tools)}): {[t.name for t in tools.tools]}')
result = await client.call_tool('test_simple_text', {})
print(f' test_simple_text: {result.content[0].text}')
resources = await client.list_resources()
print(f' Resources: {[str(r.uri) for r in resources.resources]}')
prompts = await client.list_prompts()
print(f' Prompts: {[p.name for p in prompts.prompts]}')

anyio.run(main)
"

record_http_demo "13" "simple-task"
"simple-task (HTTP)"
"Task server with long-running tool requiring task-augmented invocation."
"examples/servers/simple-task"
"uv run mcp-simple-task --port 23003"
"23003"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23003/mcp') as client:
tools = await client.list_tools()
t = tools.tools[0]
print(f' Tool: {t.name}')
print(f' task_support: {t.execution.task_support}')

anyio.run(main)
"

record_http_demo "14" "simple-task-interactive"
"simple-task-interactive (HTTP)"
"Interactive tasks: elicitation (confirm_delete) and sampling (write_haiku)."
"examples/servers/simple-task-interactive"
"uv run mcp-simple-task-interactive --port 23004"
"23004"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23004/mcp') as client:
tools = await client.list_tools()
for t in tools.tools:
print(f' Tool: {t.name} (task_support={t.execution.task_support})')

anyio.run(main)
"

record_http_demo "15" "sse-polling-demo"
"sse-polling-demo (HTTP)"
"SSE polling with periodic stream close and client auto-reconnect."
"examples/servers/sse-polling-demo"
"uv run mcp-sse-polling-demo --port 23005"
"23005"
"
import anyio
from mcp.client.client import Client

async def main():
async with Client('http://127.0.0.1:23005/mcp') as client:
tools = await client.list_tools()
print(f' Tools: {[t.name for t in tools.tools]}')
result = await client.call_tool('process_batch', {'items': 3, 'checkpoint_every': 10})
print(f' Result: {result.content[0].text}')

anyio.run(main)
"

echo ""
echo "=========================================="
echo "All recordings complete!"
echo "=========================================="
echo ""
echo "Converting to GIFs..."
for cast in evidence/[0-9].cast; do
gif="${cast%.cast}.gif"
name=$(basename "$cast" .cast)
echo -n " $name -> gif... "
agg --idle-time-limit 1.5 --speed 1.5 "$cast" "$gif" 2>&1 | tail -1
done
echo ""
echo "All done!"
ls -lh evidence/[0-9].gif

<details>

<details>
<summary>Steps to run all 15 examples</summary>

All commands assume you are in the repository root directory.

---

### 1. simple-tool (stdio)

**What it does:** Provides a `fetch` tool that fetches website content.

```bash
cd examples/servers/simple-tool

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-tool', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            tools = await session.list_tools()
            for t in tools.tools:
                print(f'Tool: {t.name} - {t.description}')

anyio.run(main)
"

Expected output:

Tool: fetch - Fetches a website and returns its content

2. simple-prompt (stdio)

What it does: Provides a simple prompt with optional context and topic arguments.

cd examples/servers/simple-prompt

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-prompt', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            prompts = await session.list_prompts()
            for p in prompts.prompts:
                print(f'Prompt: {p.name} - {p.description}')
            result = await session.get_prompt('simple', {'context': 'hello', 'topic': 'MCP'})
            for msg in result.messages:
                print(f'  [{msg.role}]: {msg.content.text}')

anyio.run(main)
"

Expected output:

Prompt: simple - A simple prompt that can take optional context and topic arguments
  [user]: Here is some relevant context: hello
  [user]: Please help me with the following topic: MCP

3. simple-resource (stdio)

What it does: Provides 3 text resources: greeting, help, about.

cd examples/servers/simple-resource

uv run python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-resource', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            resources = await session.list_resources()
            for r in resources.resources:
                result = await session.read_resource(r.uri)
                print(f'{r.name}: {result.contents[0].text}')

anyio.run(main)
"

Expected output:

greeting: Hello! This is a sample text resource.
help: This server provides a few sample text resources for testing.
about: This is the simple-resource MCP server implementation.

4. simple-pagination (stdio)

What it does: Demonstrates pagination with 25 tools (5/page), 30 resources (10/page), 20 prompts (7/page).

cd examples/servers/simple-pagination

uv run python -c "
import anyio
from mcp import types
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(command='uv', args=['run', 'mcp-simple-pagination', '--transport', 'stdio'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            all_tools, cursor, page = [], None, 0
            while True:
                p = types.PaginatedRequestParams(cursor=cursor) if cursor else None
                result = await session.list_tools(params=p)
                all_tools.extend(result.tools)
                page += 1
                print(f'Page {page}: {len(result.tools)} tools')
                if not result.next_cursor: break
                cursor = result.next_cursor
            print(f'Total: {len(all_tools)} tools across {page} pages')

anyio.run(main)
"

Expected output:

Page 1: 5 tools
Page 2: 5 tools
Page 3: 5 tools
Page 4: 5 tools
Page 5: 5 tools
Total: 25 tools across 5 pages

5. snippet: lowlevel/basic.py (stdio)

What it does: Basic low-level server with prompts using the kwargs constructor pattern.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/basic.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            prompts = await session.list_prompts()
            print(f'Prompts: {[p.name for p in prompts.prompts]}')
            result = await session.get_prompt('example-prompt', {'arg1': 'test'})
            print(f'Message: {result.messages[0].content.text}')

anyio.run(main)
"

Expected output:

Prompts: ['example-prompt']
Message: Example prompt text with argument: test

6. snippet: lowlevel/direct_call_tool_result.py (stdio)

What it does: Tool returning CallToolResult directly with structured_content and _meta.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/direct_call_tool_result.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            result = await session.call_tool('advanced_tool', {'message': 'hello'})
            print(f'Text: {result.content[0].text}')
            print(f'Structured: {result.structured_content}')

anyio.run(main)
"

Expected output:

Text: Processed: hello
Structured: {'result': 'success', 'message': 'hello'}

7. snippet: lowlevel/lifespan.py (stdio)

What it does: Server with lifespan management — connects a mock database on startup, disconnects on shutdown.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/lifespan.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            result = await session.call_tool('query_db', {'query': 'SELECT * FROM users'})
            print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output:

Result: Query results: [{'id': '1', 'name': 'Example', 'query': 'SELECT * FROM users'}]

8. snippet: lowlevel/structured_output.py (stdio)

What it does: Tool with output_schema and structured_content response.

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv', args=['run', '--frozen', 'python', 'examples/snippets/servers/lowlevel/structured_output.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            tools = await session.list_tools()
            print(f'Has output_schema: {tools.tools[0].output_schema is not None}')
            result = await session.call_tool('get_weather', {'city': 'San Francisco'})
            print(f'Structured: {result.structured_content}')

anyio.run(main)
"

Expected output:

Has output_schema: True
Structured: {'temperature': 22.5, 'condition': 'partly cloudy', 'humidity': 65, 'city': 'San Francisco'}

9. structured-output-lowlevel (stdio)

What it does: Full example server with structured output (simulated weather data).

uv run --frozen python -c "
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command='uv',
        args=['run', '--frozen', 'python',
              'examples/servers/structured-output-lowlevel/mcp_structured_output_lowlevel/__main__.py'])
    async with stdio_client(params) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            result = await session.call_tool('get_weather', {'city': 'Tokyo'})
            print(f'Structured keys: {list(result.structured_content.keys())}')

anyio.run(main)
"

Expected output:

Structured keys: ['temperature', 'conditions', 'humidity', 'wind_speed', 'timestamp']

10. simple-streamablehttp (HTTP)

What it does: Streamable HTTP server with notification stream tool and event store for resumability.

Terminal 1 — Start the server:

cd examples/servers/simple-streamablehttp
uv run mcp-simple-streamablehttp --port 3000

Wait until you see: Uvicorn running on http://127.0.0.1:3000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3000/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools: {[t.name for t in tools.tools]}')
        result = await client.call_tool('start-notification-stream', {'interval': 0.5, 'count': 3, 'caller': 'test'})
        print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output:

Tools: ['start-notification-stream']
Result: Sent 3 notifications with 0.5s interval for caller: test

11. simple-streamablehttp-stateless (HTTP)

What it does: Stateless streamable HTTP server (no session persistence between requests).

Terminal 1 — Start the server:

cd examples/servers/simple-streamablehttp-stateless
uv run mcp-simple-streamablehttp-stateless --port 3000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3000/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools: {[t.name for t in tools.tools]}')
        result = await client.call_tool('start-notification-stream', {'interval': 0.5, 'count': 3, 'caller': 'test'})
        print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output: Same as #10.

12. everything-server (HTTP)

What it does: Conformance test server implementing all MCP features — 13 tools, resources, prompts, completions, subscriptions, sampling, elicitation.

Terminal 1 — Start the server:

cd examples/servers/everything-server
uv run mcp-everything-server --port 3001

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3001/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools ({len(tools.tools)}): {[t.name for t in tools.tools]}')
        result = await client.call_tool('test_simple_text', {})
        print(f'test_simple_text: {result.content[0].text}')
        resources = await client.list_resources()
        print(f'Resources: {[str(r.uri) for r in resources.resources]}')
        prompts = await client.list_prompts()
        print(f'Prompts: {[p.name for p in prompts.prompts]}')

anyio.run(main)
"

Expected output:

Tools (13): ['test_simple_text', 'test_image_content', 'test_audio_content', ...]
test_simple_text: This is a simple text response for testing.
Resources: ['test://static-text', 'test://static-binary', 'test://watched-resource']
Prompts: ['test_simple_prompt', 'test_prompt_with_arguments', 'test_prompt_with_embedded_resource', 'test_prompt_with_image']

13. simple-task (HTTP)

What it does: Demonstrates MCP tasks with a long-running tool that requires task-augmented invocation.

Terminal 1 — Start the server:

cd examples/servers/simple-task
uv run mcp-simple-task --port 8000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:8000/mcp') as client:
        tools = await client.list_tools()
        t = tools.tools[0]
        print(f'Tool: {t.name}')
        print(f'  task_support: {t.execution.task_support}')

anyio.run(main)
"

Expected output:

Tool: long_running_task
  task_support: required

14. simple-task-interactive (HTTP)

What it does: Interactive task server demonstrating elicitation and sampling via tasks.

Terminal 1 — Start the server:

cd examples/servers/simple-task-interactive
uv run mcp-simple-task-interactive --port 8000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:8000/mcp') as client:
        tools = await client.list_tools()
        for t in tools.tools:
            print(f'Tool: {t.name} (task_support={t.execution.task_support})')

anyio.run(main)
"

Expected output:

Tool: confirm_delete (task_support=required)
Tool: write_haiku (task_support=required)

15. sse-polling-demo (HTTP)

What it does: Demonstrates SSE polling pattern with periodic stream close and client auto-reconnect.

Terminal 1 — Start the server:

cd examples/servers/sse-polling-demo
uv run mcp-sse-polling-demo --port 3000

Terminal 2 — Connect a client (from repo root):

uv run --frozen python -c "
import anyio
from mcp.client.client import Client

async def main():
    async with Client('http://127.0.0.1:3000/mcp') as client:
        tools = await client.list_tools()
        print(f'Tools: {[t.name for t in tools.tools]}')
        result = await client.call_tool('process_batch', {'items': 3, 'checkpoint_every': 10})
        print(f'Result: {result.content[0].text}')

anyio.run(main)
"

Expected output:

Tools: ['process_batch']
Result: Successfully processed 3 items with checkpoints every 10 items

1. simple-tool (stdio)

2. simple-prompt (stdio)

3. simple-resource (stdio)

4. simple-pagination (stdio)

5. snippet: basic.py (stdio)

6. snippet: direct_call_tool_result.py (stdio)

7. snippet: lifespan.py (stdio)

8. snippet: structured_output.py (stdio)

9. structured-output-lowlevel (stdio)

10. simple-streamablehttp (HTTP)

11. simple-streamablehttp-stateless (HTTP)

12. everything-server (HTTP)

13. simple-task (HTTP)

14. simple-task-interactive (HTTP)

15. sse-polling-demo (HTTP)